/*
 * main.c
 *
 *  Created on: 2009-8-7
 *      Author: wyb
 */

#include <time.h>
#include <stdio.h>
#include "xengine_mysql.h"
#include "log.h"
#include <stdlib.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <string.h>
#include <stdio.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <errno.h>
#include <stdlib.h>
#include <pthread.h>
#include "namemap.h"

extern void store();
extern void store_state(char *app_name, char *computer_name, double cpu,
		time_t current_time);
extern void
store_computer(char *computer_name, double cpu, time_t current_time);
extern void store_app(char *app_name, double cpu, time_t current_time);

#define SIZE sizeof(struct sockaddr_in)

int xsockfd, xEngineSock;

char **appNames = 0;
char *LOCAL_IP_ADDRESS = "10.2.225.199";
int LOCAL_PORT = 2346;

int numOfApp; //app的数量
int numOfPM; //物理机的数量
float* appNeedsV = 0; //每个App所需要的计算能力
int** assignM = 0; //分配矩阵，表明App是否在PM上运行，以及是否存在实体
float** stateM = 0; //状态矩阵，用来表明每个App在每个PM上的负载
int* appSizeV = 0; //每个应用的大小，用于计算部署代价
int* availableV = 0; //该机器是否处于开启状态

/*
 * 分配矩阵内存
 */
void initMatrix() {
	int i;
	assignM = malloc(numOfApp * sizeof(int*));
	for (i = 0; i < numOfApp; i++) {
		assignM[i] = malloc(numOfPM * sizeof(int));
	}

	stateM = malloc(numOfApp * sizeof(float*));
	for (i = 0; i < numOfApp; i++) {
		stateM[i] = malloc(numOfPM * sizeof(float));
	}

	appNeedsV = malloc(numOfApp * sizeof(float));
	availableV = malloc(numOfPM * sizeof(int));
	memset(availableV, 1, sizeof(int) * numOfPM);
	appSizeV = malloc(numOfApp * sizeof(int));
}

/*
 * 回收空间
 */
void destoryMatrix() {
	int i;
	if (assignM != 0) {
		for (i = 0; i < numOfApp; i++) {
			free(assignM[i]);
		}
		free(assignM);
	}

	if (stateM != 0) {
		for (i = 0; i < numOfApp; i++) {
			free(stateM[i]);
		}
		free(stateM);
	}
	if (appNeedsV != 0)
		free(appNeedsV);
	if (availableV != 0)
		free(availableV);
	if (appSizeV != 0)
		free(appSizeV);
}

/*
 * 根据状态矩阵反推出app的需要
 */
void calculateAppNeeds() {
	int i, j;
	float sum;
	for (i = 0; i < numOfApp; i++) {
		sum = 0;
		for (j = 0; j < numOfPM; j++) {
			if (availableV[j]) {
				sum += stateM[i][j];
			}
		}
		appNeedsV[i] = sum;
	}
}

/*
 * 创建socket连接
 */
void createSocket(int *sockfd, const char* ip, int port) {
	struct sockaddr_in server = { AF_INET, htons(port), { INADDR_ANY } };
	int optval = 1;

	server.sin_addr.s_addr = inet_addr(ip);
	if ((*sockfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
		perror("socket call failed");
		exit(1);
	}
	if (setsockopt(*sockfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval))
			< 0) {
		perror("socket reuse");
		close(*sockfd);
		exit(1);
	}
	if (bind(*sockfd, (struct sockaddr *) &server, SIZE) == -1) {
		//绑定
		perror("bind call error");
		close(*sockfd);
		exit(1);
	}
}

/*
 * 从字符串中获取获取名称为paraName的value
 * 参数：offset，字符串的偏移量
 *       result即名称为paraName的value，用于返回
 * 返回：处理完之后新的offset
 */
int getStringParameter(char* msg, int offset, const char* paraName,
		char* result) {
	int i, p;
	i = 0;
	while (paraName[i] != 0) {
		if (paraName[i] != msg[offset + i]) {
			return -1;
		}
		i++;
	}
	i++;
	p = 0;
	while (msg[offset + i] != '&' && msg[offset + i] != '\0' && msg[offset + i]
			!= '\r' && msg[offset + i] != '\n') {
		result[p] = msg[offset + i];
		i++;
		p++;
	}
	result[p] = '\0';
	return offset + i + 1;
}

/*
 * 从字符串中获取获取名称为paraName的value，并转成整形输出
 * 参数：offset，字符串的偏移量
 *       result即名称为paraName的value，用于返回
 * 返回：处理完之后新的offset
 */
int getIntParameter(char* msg, int offset, const char* paraName, int *result) {
	char temp[128];
	int ret;
	ret = getStringParameter(msg, offset, paraName, temp);
	*result = atoi(temp);
	return ret;
}

/*
 * 将字符串包装成Erlang可识别的格式
 * 参数：msg 欲包装的字符串
 *       bytes 包装完后的包
 * 返回：包装完后的数据长度
 */
int packetErlangData(const char* msg, char* bytes) {
	printf("SEND MESSAGE: %s \n", msg);
	int l = strlen(msg);
	bytes[0] = '\0';
	bytes[1] = '\0';
	bytes[2] = '\0';
	bytes[3] = (char) l;
	memcpy(&bytes[4], msg, l);
	return l + 4;
}

/*
 * 接受Erlang传来的数据，并将数据转成C可是别的格式
 * 参数：rev 用于返回收到的数据
 */
int receive(int sockfd, char * rev) {
	char temp[8];
	int p = 0;
	int ret;
	printf("try to receive data\n");
	do {
		ret = recv(sockfd, &rev[p], 1, 0);
		if (ret != 1) {
			printf("client closed\n");
			return -1;
		}
		if (rev[p] == '\0') {
			ret = recv(sockfd, temp, 3, 0);
			if (ret != 3) {
				printf("client closed\n");
				return -1;
			}
		} else {
			p++;
		}
	} while (rev[p - 1] != '\n');
	rev[p] = '\0';

	printf("RECEIVE MESSAGE: %s \n", rev);
	return 0;
}

/*
 * 处理请求App状态的返回信息
 */
void recvMsgType2(int appId, char * rev) {
	int i;
	int msgType, machineNum;
	int offset = 0;
	char machineName[64], appName[64];
	int machineId, state, intLoad;
	float load;
	offset = getIntParameter(rev, offset, "msgtype", &msgType);
	offset = getIntParameter(rev, offset, "appid", &appId);
	offset = getStringParameter(rev, offset, "appname", appName);
	offset = getIntParameter(rev, offset, "machinenum", &machineNum);
	appId -= 1;
	strcpy(appNames[appId], appName);
	for (i = 0; i < machineNum; i++) {
		offset = getStringParameter(rev, offset, "machinename", machineName);
		offset = getIntParameter(rev, offset, "state", &state);
		offset = getIntParameter(rev, offset, "load", &intLoad);
		machineId = nameToId(machineName);
		load = (float) intLoad / 100;
		stateM[appId][machineId] = load;
		assignM[appId][machineId] = state - 1;
	}
	appSizeV[appId] = 10;

}
/*
 * 算法模块请求每个App的状态
 * 参数：appId 所请求的App的编号
 */
void sendMsgType2(int appId) {
	char str[255];
	char sev[255];
	char rev[4096];
	int l;
	if (appId < 0) {
		sprintf(str, "msgtype=2&response=finish");
		l = packetErlangData(str, sev);
		send(xEngineSock, sev, l, 0);
	} else {
		sprintf(str, "msgtype=2&appid=%d\r\n", appId + 1);
		l = packetErlangData(str, sev);
		send(xEngineSock, sev, l, 0);
		receive(xEngineSock, rev);
		recvMsgType2(appId, rev);
	}
}

/*
 * 调度模块向算法模块发起调度请求，处理这部分内容
 * 参数：msg 所收的请求字符串
 *      offset 字符串去除前面已经处理过的字符的偏移量
 */
void processMsgType1(char* msg, int offset) {
	int i, l;
	char sev[255];

	printf("free space\n");
	destoryMatrix();
	if (appNames) {
		for (i = 0; i < numOfApp; i++) {
			free(appNames[i]);
		}
		free(appNames);
	}
	printf("malloc space\n");
	offset = getIntParameter(msg, offset, "machinenum", &numOfPM);
	offset = getIntParameter(msg, offset, "appnum", &numOfApp);
	initMatrix();
	printf("finish malloc\n");
	appNames = malloc(numOfApp * sizeof(char*));
	for (i = 0; i < numOfApp; i++) {
		appNames[i] = malloc(64 * sizeof(char));
	}

	l = packetErlangData("msgtype=1&response=ok\r\n", sev);
	send(xEngineSock, sev, l, 0);

	clearNamemap();
	for (i = 0; i < numOfApp; i++) {
		sendMsgType2(i);
	}
	sendMsgType2(-1);

	calculateAppNeeds();
	//TODO:存入获取的数据
	//stateM[appId][PMId]
	//appNeedsV[appId]
	//appName = appNames[appId]
	//PMName = idToName(PMId);
	//PMId = nameToid(PMName);
	store();

}

/*
 * 收到了请求字符串，按照字符串的msgtype开始处理
 * 参数：msg 所收的请求字符串
 */
void processMessage(char* msg) {
	int offset = 0;
	int msgtype;
	offset = getIntParameter(msg, offset, "msgtype", &msgtype);
	switch (msgtype) {
	case 1:
		processMsgType1(msg, offset);
		printf("finish process MESSAGE\n");
		break;
	}
}

/*
 * 开始监听端口
 */
void listenport(int *sockfd, int *newsockfd) {
	char rev[4096];
	int ret;
	ret = 0;
	while (1) {
		if (listen(*sockfd, 5) == -1) {
			perror("listen call error");
			close(xsockfd);
			exit(1);
		}
		printf("forecast-xengine start listen....\n");
		if ((*newsockfd = accept(*sockfd, NULL, NULL)) == -1) {
			perror("accept call error");
			continue;
		} else {
			printf("connected...\n");
			do {
				ret = receive(*newsockfd, rev);
				if (ret == 0)
					processMessage(rev);
			} while (ret == 0);
			close(*newsockfd);
			printf("disconnected...\n");
		}
	}
}

void store() {
	time_t current_time;
	if (time(&current_time) != -1) {
		xengine_mysql_init(URL, DB, USERNAME, PASSWORD);

		int i, j;
		for (i = 0; i < numOfApp; i++) {
			store_app(appNames[i], appNeedsV[i], current_time);
		}

		char t[256];
		for (i = 0; i < numOfPM; i++) {
			double temp = 0.0;
			for (j = 0; j < numOfApp; j++) {
				temp += stateM[j][i];
			}
			idToName(i, t);
			store_computer(t, temp, current_time);
		}

		for (i = 0; i < numOfApp; i++) {
			for (j = 0; j < numOfPM; j++) {
				idToName(j, t);
				store_state(appNames[i], t, stateM[i][j], current_time);
			}
		}

		xengine_mysql_close();
	} else {
		printf("time error!\n");
	}
}

void store_computer(char *computer_name, double cpu, time_t current_time) {

	struct computer *computer = get_computer(computer_name);

	if (computer == NULL) {
		insert_computer(computer_name);
		computer = get_computer(computer_name);
	}

	insert_computer_history(computer->id, DEFAULT_FIELD, cpu, current_time);

	free(computer);
}

void store_state(char *app_name, char *computer_name, double cpu,
		time_t current_time) {

	struct app *app = get_app(app_name);
	struct computer *computer = get_computer(computer_name);

	if (app == NULL) {
		insert_app(app_name);
		app = get_app(app_name);
	}
	if (computer == NULL) {
		insert_computer(computer_name);
		computer = get_computer(computer_name);
	}

	insert_computer_app(app->id, computer->id, DEFAULT_FIELD, cpu, current_time);

	free(app);
	free(computer);
}

void store_app(char *app_name, double cpu, time_t current_time) {

	struct app *app = get_app(app_name);

	if (app == NULL) {
		insert_app(app_name);
		app = get_app(app_name);
	}

	insert_app_history(app->id, DEFAULT_FIELD, cpu, current_time);

	free(app);
}

/*
 * 主函数，创建连接并监听
 */
int main(int argc, char** argv) {
	set_project_name("xengine_history");

	createSocket(&xsockfd, LOCAL_IP_ADDRESS, LOCAL_PORT);
	listenport(&xsockfd, &xEngineSock);
	return (EXIT_SUCCESS);
}
